package com.mapbased.sfw.store;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import com.mapbased.sfw.common.HbaseClient;

public class HbaseSiteStore extends SiteStore {

	public static final String HBASE_TABLE_NAME = "site";
	private static final byte[] family = "x".getBytes();

	public HbaseSiteStore(String id) {
		super(id);
	}
	public byte[] getData(final String path,final PathDataType type) {
		try {
		return (byte[])	HbaseClient.INSTANCE.getData(HBASE_TABLE_NAME,
					new HbaseClient.DataGetter() {

						@Override
						public Object getData(HTable table) throws IOException {
							Get get = new Get(Bytes.toBytes(siteId));

							 
							get.addColumn(family,  Bytes
									.toBytes(type.getName()+path));
							get.setMaxVersions(1);

							Result r = table.get(get);
							
						 
							if (!r.isEmpty()) {
								
							   r.raw()[0].getValue();
							}

							return null;
						}

					});
		} catch (IOException e) {
			throw new java.lang.RuntimeException(e);
		}
		 
	}
	@Override
	public void delete(final String path, final PathDataType type) {
		try {
			HbaseClient.INSTANCE.getData(HBASE_TABLE_NAME,
					new HbaseClient.DataGetter() {

						@Override
						public Object getData(HTable table) throws IOException {
							Delete del = new Delete(Bytes.toBytes(siteId));

							del.deleteColumn(family, Bytes
									.toBytes(type.getName()+path));

							table.delete(del);

							return null;
						}

					});
		} catch (IOException e) {
			throw new java.lang.RuntimeException(e);
		}

	}

	private static String encodePath(PathData path) {
		return path.type.getName()+path.path;
	}

	static private PathData decodePath(String encodePath) {
		if (encodePath != null  ) {
			PathData pd=new PathData();
			pd.type=PathDataType.fromString(encodePath);
			if(pd.type!=null){
				pd.path=encodePath.substring(2);
			}
		}
		return null;
	}

	 

	@Override
	public void put(final PathData kv) {
		try {
			HbaseClient.INSTANCE.getData(HBASE_TABLE_NAME,
					new HbaseClient.DataGetter() {

						@Override
						public Object getData(HTable table) throws IOException {
							Put p = new Put(Bytes.toBytes(siteId));
							p.add(family, Bytes.toBytes(encodePath(kv)),
									System.currentTimeMillis(), kv.value );
							table.put(p);
							return null;
						}

					});
		} catch (IOException e) {
			throw new java.lang.RuntimeException(e);
		}

	}

	@Override
	public boolean exist() {
		try {
			return (Boolean) HbaseClient.INSTANCE.getData(HBASE_TABLE_NAME,
					new HbaseClient.DataGetter() {

						@Override
						public Object getData(HTable table) throws IOException {
							Get get = new Get(Bytes.toBytes(siteId));

							get.addFamily(family);
							get.setMaxVersions(1);
							return table.exists(get);

						}

					});
		} catch (IOException e) {
			throw new java.lang.RuntimeException(e);
		}

	}

	@Override
	public List<PathData> getAllPathData() {
		final List<PathData> l = new java.util.ArrayList<PathData>();
		try {
			HbaseClient.INSTANCE.getData(HBASE_TABLE_NAME,
					new HbaseClient.DataGetter() {

						@Override
						public Object getData(HTable table) throws IOException {
							Get get = new Get(Bytes.toBytes(siteId));

							get.addFamily(family);
							get.setMaxVersions(1);

							Result r = table.get(get);
							
						 
							if (!r.isEmpty()) {
								
								NavigableMap<byte[],NavigableMap<Long,byte[]>> cols= r.getMap().get(family);
								

								 
								
								for (Map.Entry<byte[],NavigableMap<Long,byte[]>> me : cols
										.entrySet()) {

									String key = Bytes.toString(me.getKey());
									Map.Entry< Long,byte[]> cel=me.getValue().lastEntry();
								 
									PathData pd= decodePath(key);
									if(pd.type!=null){
										
									
										pd.lastModified=cel.getKey();
	
										pd.value = cel.getValue();
										if ( pd.path.length()>0) {
											l.add(pd);
										}
									}

								}

							}

							return l;
						}

					});
		} catch (IOException e) {
			throw new java.lang.RuntimeException(e);
		}
		return l;
	}

}
